package hk.uic.dempsey;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.*;
import org.tartarus.snowball.ext.PorterStemmer;

import java.io.IOException;

public class Task2 {

   public static class MapperForWord extends Mapper<LongWritable,Text,Text,LongWritable>{
       private final static LongWritable one= new LongWritable(1);
       //private Text word=new Text();
       public String modify(String str)
       {
           String temp = str.replaceAll("//W", "");  // //W means the non-character string
           return temp;
       }
       public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
          String line=value.toString();
          String[] words=line.split(" ");
          for(String word:words) {
              String temp;
              word=word.toLowerCase();
              temp=modify(word);
              temp=temp.replaceAll("[-_=|:;'`!?,.\"?~#%*&(){}]",""); //keep eliminate the special symbol
            //  PorterStemmer porterStemmer = new PorterStemmer();
            //  porterStemmer.setCurrent(temp);
            //  temp = porterStemmer.getCurrent();
              context.write(new Text(temp), one);
          }
   }
       }


   public static class SumForKey extends Reducer<Text,LongWritable,Text,LongWritable>{
       private LongWritable sumAll=new LongWritable();
       public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException,InterruptedException{
           long sum=0;
           for(LongWritable value:values){
               sum=sum+value.get();          //calculate the sum for all
           }
           sumAll.set(sum);
           context.write(key,sumAll);
       }
   }

   public static void main(String[] args) throws Exception{
       long startTime = System.currentTimeMillis();
       //_____________________________clock start
       Configuration conf = new Configuration();
       Job job= Job.getInstance(conf,"word count");  //set the job
       job.setJarByClass(Task2.class);
       FileInputFormat.addInputPath(job,new Path(args[0]));

       Path outputPath = new Path(args[1]);
       FileSystem fileSystem =FileSystem.get(conf);
       if (fileSystem.exists(outputPath)){
           fileSystem.delete(outputPath,true);
           System.out.println("outputPath: "+args[1]+" exists, but has been deleted.");
       }   //set the output path
       //Mapper information
       job.setMapperClass(MapperForWord.class);
       job.setMapOutputKeyClass(Text.class);
       job.setOutputValueClass(LongWritable.class);        //set the mapper

       //Reducer information
       job.setReducerClass(SumForKey.class);
       job.setOutputKeyClass(Text.class);   //set the reducer
       job.setOutputValueClass(LongWritable.class);

       FileOutputFormat.setOutputPath(job,new Path(args[1]));   //set the output file path.
       if(job.waitForCompletion(true)) {
           long endTime = System.currentTimeMillis();
           System.out.println("---------------------Time use:"+(endTime-startTime)+"ms");
           System.exit(job.waitForCompletion(true) ? 0 : 1);
       }
   }

}
